package com.kylin.common.kafka.logback;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * 自定义KafkaAppender将日志输出到Kafka中
 * @author liuyazhuang
 *
 */
@Component
@Slf4j
public class KafkaAppender extends AppenderBase<ILoggingEvent> {

    Producer<String, Object> producer;
    private String zookeeperHost;
    private String brokerList;
    private String topic;
    private Formatter formatter;
    private boolean enable;

    @Override
    public void start() {
        if(this.formatter == null){
            this.formatter = new MessageFormatter();
        }
        System.out.println("============>"+(this.enable == false));
        if(this.enable) {
            super.start();
            this.producer = new KafkaProducer<String, Object>(getProperties());
        }
    }

    @Override
    public void stop() {
        super.stop();
        if(this.producer != null) {
            this.producer.close();
        }
    }

    @Override
    protected void append(ILoggingEvent event) {
        String payload = this.formatter.format(event);
        if(this.producer != null) {
            // 往kafka发送消息
            this.producer.send(new ProducerRecord(this.topic, payload), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println(1111);
                    System.out.println(e);
                    System.out.println();
                }
            });
        }
    }


    public Properties getProperties(){
        Properties props = new Properties();
        props.put("bootstrap.servers", this.brokerList);
        //props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "112.74.55.160:9092");
        // 当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性的级别,分别是0, 1，all。
//        props.put("acks", "all");
        //props.put(ProducerConfig.ACKS_CONFIG, "all");
        // 请求失败，生产者会自动重试，指定是0次，如果启用重试，则会有重复消息的可能性
//        props.put("retries", 0);
        //props.put(ProducerConfig.RETRIES_CONFIG, 0);
        // 生产者缓存每个分区未发送的消息,缓存的大小是通过 batch.size 配置指定的，默认值是16KB
        props.put("batch.size", 16384);
        /**
         * 默认值就是0，消息是立刻发送的，即便batch.size缓冲空间还没有满
         * 如果想减少请求的数量，可以设置 linger.ms 大于0，即消息在缓冲区保留的时间，超过设置的值就会被提交到          服务端
         * 通俗解释是，本该早就发出去的消息被迫至少等待了linger.ms时间，相对于这时间内积累了更多消息，批量发送           减少请求
         * 如果batch被填满或者linger.ms达到上限，满足其中一个就会被发送
         */
        props.put("linger.ms", 1);
        /**
         * buffer.memory的用来约束Kafka Producer能够使用的内存缓冲的大小的，默认值32MB。
         * 如果buffer.memory设置的太小，可能导致消息快速的写入内存缓冲里，但Sender线程来不及把消息发送到             Kafka服务器
         * 会造成内存缓冲很快就被写满，而一旦被写满，就会阻塞用户线程，不让继续往Kafka写消息了
         * buffer.memory要大于batch.size，否则会报申请内存不#足的错误，不要超过物理内存，根据实际情况调整
         * 需要结合实际业务情况压测进行配置
         */
        props.put("buffer.memory", 33554432);
        /**
         * key的序列化器，将用户提供的 key和value对象ProducerRecord 进行序列化处理，key.serializer必须被          设置，
         * 即使消息中没有指定key，序列化器必须是一个实
         org.apache.kafka.common.serialization.Serializer接口的类，
         * 将key序列化成字节数组。
         */
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }


    public Formatter getFormatter() {
        return formatter;
    }

    public void setFormatter(Formatter formatter) {
        this.formatter = formatter;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getZookeeperHost() {
        return zookeeperHost;
    }

    public void setZookeeperHost(String zookeeperHost) {
        this.zookeeperHost = zookeeperHost;
    }

    public String getBrokerList() {
        return brokerList;
    }

    public void setBrokerList(String brokerList) {
        this.brokerList = brokerList;
    }

    public boolean isEnable() {
        return enable;
    }

    public void setEnable(boolean enable) {
        this.enable = enable;
    }
}
